package com.code.soulnetty.modbus.client.channel.impl;

import com.code.soulnetty.modbus.client.message.ProtocolRequest;
import com.code.soulnetty.modbus.client.message.ProtocolResponse;
import com.code.soulnetty.modbus.client.service.ProtocolResponseService;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;

import java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author lyf
 * @date 2022-12-17 14:35
 */
@Slf4j
public class NettySyncChannel extends NettyAsyncChannel {

    private ReentrantLock lock = new ReentrantLock();

    private Condition condition = lock.newCondition();

    public NettySyncChannel(SocketAddress remoteSocketAddress, Channel channel, ProtocolResponseService responseService, int retryCount, long timeout, long reqIntl) {
        super(remoteSocketAddress, channel, responseService, retryCount, timeout, reqIntl);
    }

    @Override
    public CompletableFuture<ProtocolResponse> send(ProtocolRequest request) throws Exception {
        long start = System.currentTimeMillis();
        final Pair<Boolean, String> msgIdInfo = request.msgIdInfo();
        String msgId = msgIdInfo.getRight();
        CompletableFuture<ProtocolResponse> future = super.send(request);
        lock.lock();
        try {
            long totalTimeout = getRetryCount() * getTimeout();
            while (getRequestMap().get(msgId) != null) {
                try {
                    boolean receive = condition.await(100, TimeUnit.MILLISECONDS);
                    if (receive || (System.currentTimeMillis() - start) > totalTimeout || !isConnected()) {
                        break;
                    }
                } catch (InterruptedException e) {
                    log.error("send data error {}", getRemoteAddress(), e);
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        } finally {
            lock.unlock();
        }
        return future;
    }

    @Override
    public void received(ProtocolResponse response) {
        lock.lock();
        try {
            // 当收到netty服务端数据后，唤醒所有等待响应结果的发送线程，如 @send 方法
            condition.signalAll();
        } finally {
            lock.unlock();
        }
        super.received(response);
    }

    @Override
    public ProtocolRequest request(String msgId) {
        return getRequestMap().get(msgId);
    }
}
